package cn.myframe;

import cn.myframe.consumer.MessageEvent;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.EnableKafka;

/**
 * 
 * 启动方法
 * 
 * @author  ynz
 * @email   ynz@rojao.cn
 * @version 创建时间：2018年6月25日 下午5:32:00
 */
@SpringBootApplication
@EnableKafka
public class KafkaConsumerApplication {
	
	public static void main(String[] args){
        SpringApplication app = new SpringApplication(KafkaConsumerApplication.class);
       // app.setWebEnvironment(false);
        app.run(args);
    }


    @Bean
    NewTopic topic(){
	    return new NewTopic("confirm-topic",1,(short)1);
    }

    @Bean
    public KStream<String, MessageEvent> kStream(StreamsBuilder streamsBuilder){
        KStream<String,MessageEvent> stream= streamsBuilder.stream("my-topic");
        stream.map((key,value)->{
            value.setName(value.getName().toUpperCase());
            return new KeyValue<>(key,value);
        }).to("anorther-topic");
        return stream;
    }

}
